//! 集群协议解析器
//! 
//! 处理CLUSTER命令的解析和RESP协议转换

use crate::cluster::commands::{RedisClusterCommand, ClusterCommandResult};
use crate::cluster::{ClusterError, ClusterResult};

/// CLUSTER命令解析器
pub struct ClusterProtocolParser;

impl ClusterProtocolParser {
    /// 解析CLUSTER命令
    pub fn parse_cluster_command(args: &[String]) -> ClusterResult<RedisClusterCommand> {
        if args.is_empty() {
            return Err(ClusterError::Config("CLUSTER command requires subcommand".to_string()));
        }
        
        let subcommand = args[0].to_uppercase();
        
        match subcommand.as_str() {
            "MEET" => {
                if args.len() != 3 {
                    return Err(ClusterError::Config("CLUSTER MEET requires ip and port".to_string()));
                }
                let ip = args[1].clone();
                let port = args[2].parse::<u16>()
                    .map_err(|_| ClusterError::Config("Invalid port number".to_string()))?;
                Ok(RedisClusterCommand::Meet { ip, port })
            }
            
            "FORGET" => {
                if args.len() != 2 {
                    return Err(ClusterError::Config("CLUSTER FORGET requires node id".to_string()));
                }
                let node_id = args[1].clone();
                Ok(RedisClusterCommand::Forget { node_id })
            }
            
            "NODES" => {
                if args.len() != 1 {
                    return Err(ClusterError::Config("CLUSTER NODES takes no arguments".to_string()));
                }
                Ok(RedisClusterCommand::Nodes)
            }
            
            "INFO" => {
                if args.len() != 1 {
                    return Err(ClusterError::Config("CLUSTER INFO takes no arguments".to_string()));
                }
                Ok(RedisClusterCommand::Info)
            }
            
            "SLOTS" => {
                if args.len() != 1 {
                    return Err(ClusterError::Config("CLUSTER SLOTS takes no arguments".to_string()));
                }
                Ok(RedisClusterCommand::Slots)
            }
            
            "KEYSLOT" => {
                if args.len() != 2 {
                    return Err(ClusterError::Config("CLUSTER KEYSLOT requires key".to_string()));
                }
                let key = args[1].clone();
                Ok(RedisClusterCommand::KeySlot { key })
            }
            
            "ADDSLOTS" => {
                if args.len() < 2 {
                    return Err(ClusterError::Config("CLUSTER ADDSLOTS requires slot numbers".to_string()));
                }
                let mut slots = Vec::new();
                for arg in &args[1..] {
                    let slot = arg.parse::<u16>()
                        .map_err(|_| ClusterError::Config(format!("Invalid slot number: {}", arg)))?;
                    if slot >= 16384 {
                        return Err(ClusterError::Config(format!("Slot number out of range: {}", slot)));
                    }
                    slots.push(slot);
                }
                Ok(RedisClusterCommand::AddSlots { slots })
            }
            
            "DELSLOTS" => {
                if args.len() < 2 {
                    return Err(ClusterError::Config("CLUSTER DELSLOTS requires slot numbers".to_string()));
                }
                let mut slots = Vec::new();
                for arg in &args[1..] {
                    let slot = arg.parse::<u16>()
                        .map_err(|_| ClusterError::Config(format!("Invalid slot number: {}", arg)))?;
                    if slot >= 16384 {
                        return Err(ClusterError::Config(format!("Slot number out of range: {}", slot)));
                    }
                    slots.push(slot);
                }
                Ok(RedisClusterCommand::DelSlots { slots })
            }
            
            "FLUSHSLOTS" => {
                if args.len() != 1 {
                    return Err(ClusterError::Config("CLUSTER FLUSHSLOTS takes no arguments".to_string()));
                }
                Ok(RedisClusterCommand::FlushSlots)
            }
            
            "SETSLOT" => {
                if args.len() < 3 {
                    return Err(ClusterError::Config("CLUSTER SETSLOT requires slot and subcommand".to_string()));
                }
                let slot = args[1].parse::<u16>()
                    .map_err(|_| ClusterError::Config("Invalid slot number".to_string()))?;
                if slot >= 16384 {
                    return Err(ClusterError::Config(format!("Slot number out of range: {}", slot)));
                }
                let subcommand = args[2].clone();
                let node_id = if args.len() > 3 { Some(args[3].clone()) } else { None };
                Ok(RedisClusterCommand::SetSlot { slot, subcommand, node_id })
            }
            
            "MYID" => {
                if args.len() != 1 {
                    return Err(ClusterError::Config("CLUSTER MYID takes no arguments".to_string()));
                }
                Ok(RedisClusterCommand::MyId)
            }
            
            "RESET" => {
                let option = if args.len() > 1 { Some(args[1].clone()) } else { None };
                Ok(RedisClusterCommand::Reset { option })
            }
            
            "SAVECONFIG" => {
                if args.len() != 1 {
                    return Err(ClusterError::Config("CLUSTER SAVECONFIG takes no arguments".to_string()));
                }
                Ok(RedisClusterCommand::SaveConfig)
            }
            
            "REPLICATE" => {
                if args.len() != 2 {
                    return Err(ClusterError::Config("CLUSTER REPLICATE requires master node id".to_string()));
                }
                let node_id = args[1].clone();
                Ok(RedisClusterCommand::Replicate { node_id })
            }
            
            "FAILOVER" => {
                let option = if args.len() > 1 { Some(args[1].clone()) } else { None };
                Ok(RedisClusterCommand::Failover { option })
            }
            
            "COUNTKEYSINSLOT" => {
                if args.len() != 2 {
                    return Err(ClusterError::Config("CLUSTER COUNTKEYSINSLOT requires slot number".to_string()));
                }
                let slot = args[1].parse::<u16>()
                    .map_err(|_| ClusterError::Config("Invalid slot number".to_string()))?;
                Ok(RedisClusterCommand::CountKeysInSlot { slot })
            }
            
            "GETKEYSINSLOT" => {
                if args.len() != 3 {
                    return Err(ClusterError::Config("CLUSTER GETKEYSINSLOT requires slot and count".to_string()));
                }
                let slot = args[1].parse::<u16>()
                    .map_err(|_| ClusterError::Config("Invalid slot number".to_string()))?;
                let count = args[2].parse::<u32>()
                    .map_err(|_| ClusterError::Config("Invalid count".to_string()))?;
                Ok(RedisClusterCommand::GetKeysInSlot { slot, count })
            }
            
            _ => Err(ClusterError::Config(format!("Unknown CLUSTER subcommand: {}", subcommand)))
        }
    }
    
    /// 将集群命令结果转换为RESP协议格式
    pub fn format_cluster_response(result: &ClusterCommandResult) -> String {
        match result {
            ClusterCommandResult::SimpleString(s) => format!("+{}\r\n", s),
            ClusterCommandResult::BulkString(s) => {
                if s.is_empty() {
                    "$0\r\n\r\n".to_string()
                } else {
                    format!("${}\r\n{}\r\n", s.len(), s)
                }
            }
            ClusterCommandResult::Integer(i) => format!(":{}\r\n", i),
            ClusterCommandResult::Array(arr) => {
                let mut response = format!("*{}\r\n", arr.len());
                for item in arr {
                    response.push_str(&format!("${}\r\n{}\r\n", item.len(), item));
                }
                response
            }
            ClusterCommandResult::Error(e) => format!("-ERR {}\r\n", e),
            ClusterCommandResult::Null => "$-1\r\n".to_string(),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    
    #[test]
    fn test_parse_cluster_nodes() {
        let args = vec!["NODES".to_string()];
        let cmd = ClusterProtocolParser::parse_cluster_command(&args).unwrap();
        assert!(matches!(cmd, RedisClusterCommand::Nodes));
    }
    
    #[test]
    fn test_parse_cluster_info() {
        let args = vec!["INFO".to_string()];
        let cmd = ClusterProtocolParser::parse_cluster_command(&args).unwrap();
        assert!(matches!(cmd, RedisClusterCommand::Info));
    }
    
    #[test]
    fn test_parse_cluster_meet() {
        let args = vec!["MEET".to_string(), "127.0.0.1".to_string(), "7000".to_string()];
        let cmd = ClusterProtocolParser::parse_cluster_command(&args).unwrap();
        match cmd {
            RedisClusterCommand::Meet { ip, port } => {
                assert_eq!(ip, "127.0.0.1");
                assert_eq!(port, 7000);
            }
            _ => panic!("Expected Meet command"),
        }
    }
    
    #[test]
    fn test_parse_cluster_keyslot() {
        let args = vec!["KEYSLOT".to_string(), "testkey".to_string()];
        let cmd = ClusterProtocolParser::parse_cluster_command(&args).unwrap();
        match cmd {
            RedisClusterCommand::KeySlot { key } => {
                assert_eq!(key, "testkey");
            }
            _ => panic!("Expected KeySlot command"),
        }
    }
    
    #[test]
    fn test_format_simple_string() {
        let result = ClusterCommandResult::SimpleString("OK".to_string());
        let response = ClusterProtocolParser::format_cluster_response(&result);
        assert_eq!(response, "+OK\r\n");
    }
    
    #[test]
    fn test_format_bulk_string() {
        let result = ClusterCommandResult::BulkString("hello world".to_string());
        let response = ClusterProtocolParser::format_cluster_response(&result);
        assert_eq!(response, "$11\r\nhello world\r\n");
    }
    
    #[test]
    fn test_format_integer() {
        let result = ClusterCommandResult::Integer(42);
        let response = ClusterProtocolParser::format_cluster_response(&result);
        assert_eq!(response, ":42\r\n");
    }
    
    #[test]
    fn test_format_error() {
        let result = ClusterCommandResult::Error("Something went wrong".to_string());
        let response = ClusterProtocolParser::format_cluster_response(&result);
        assert_eq!(response, "-ERR Something went wrong\r\n");
    }
    
    #[test]
    fn test_format_array() {
        let result = ClusterCommandResult::Array(vec![
            "item1".to_string(),
            "item2".to_string(),
        ]);
        let response = ClusterProtocolParser::format_cluster_response(&result);
        assert_eq!(response, "*2\r\n$5\r\nitem1\r\n$5\r\nitem2\r\n");
    }
    
    #[test]
    fn test_invalid_slot_number() {
        let args = vec!["KEYSLOT".to_string()]; // Missing key
        let result = ClusterProtocolParser::parse_cluster_command(&args);
        assert!(result.is_err());
    }
    
    #[test]
    fn test_unknown_subcommand() {
        let args = vec!["UNKNOWN".to_string()];
        let result = ClusterProtocolParser::parse_cluster_command(&args);
        assert!(result.is_err());
    }
}